package com.atguigu.flink.demo04;

import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.source.SocketTextStreamFunction;

/**
 * @author admin
 * @date 2021/8/10
 */
public class MysqlSink {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        /**
         * 从网络中读取数据
         * 参数：ip,段口，分隔符，重试次数
         */
        SocketTextStreamFunction socketText = new SocketTextStreamFunction("hadoop102", 9999, "\n", 3);
        DataStreamSource<String> source = env.addSource(socketText);


        // 将数据转换成对对象
        SingleOutputStreamOperator<UserBean> map = source.map(e -> new UserBean(e.split(",")));



        // 执行插入操作
        JdbcStatementBuilder<UserBean> builder = (JdbcStatementBuilder<UserBean>) (statement, userBean) -> {
            statement.setInt(1, userBean.getId());
            statement.setString(2, userBean.getName());
            statement.setInt(3, userBean.getAge());
            statement.setString(4, userBean.getSex());
        };

        // 执行配置
        JdbcExecutionOptions executionOptions = new JdbcExecutionOptions.Builder()
                .withBatchSize(1) // 测试，所以设置为1
                .withBatchIntervalMs(200)
                .withMaxRetries(5)
                .build();

        // 连接配置
        JdbcConnectionOptions connectionOptions = new JdbcConnectionOptions
                .JdbcConnectionOptionsBuilder()
                .withUrl("jdbc:mysql://hadoop102:3306/demo?useSSL=false") // url
                .withDriverName("com.mysql.jdbc.Driver") // driver
                .withUsername("root") // 账号
                .withPassword("123321") // 密码
                .withConnectionCheckTimeoutSeconds(30) // 连接超时时间 单位：秒
                .build();

        // 插入的sql语句
        String insertSql = "insert into t_user(id,name,age,sex) values(?,?,?,?)";

        SinkFunction<UserBean> sink = JdbcSink.sink(insertSql,builder,executionOptions,connectionOptions);
        map.print();
        map.addSink(sink);

        env.execute();

    }


}
